package com.gitee.melin.bee.core.jdbc.dialect.impl;

import com.gitee.melin.bee.core.jdbc.dialect.AbstractJdbcDialect;
import com.gitee.melin.bee.core.jdbc.enums.DataSourceType;
import com.gitee.melin.bee.core.jdbc.relational.BeeTableType;
import com.gitee.melin.bee.core.jdbc.relational.ConnectionInfo;
import com.gitee.melin.bee.core.jdbc.relational.MetaColumn;
import com.gitee.melin.bee.core.jdbc.relational.MetaTable;
import com.google.common.collect.Lists;
import java.sql.*;
import java.util.List;
import java.util.Locale;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RedshiftJdbcDialect extends AbstractJdbcDialect {

    private static final Logger LOG = LoggerFactory.getLogger(RedshiftJdbcDialect.class);

    @Override
    public DataSourceType getDataSourceType() {
        return DataSourceType.POSTGRESQL;
    }

    private static final String QUERY_DATABASES_SQL =
            "select datname from pg_catalog.pg_database where datistemplate = false";

    private static final String[] SYS_SCHEMA_NAMES =
            new String[] {"pg_catalog", "pg_internal", "pg_s3", "pg_automv", "pg_auto_copy"};

    public RedshiftJdbcDialect(ConnectionInfo connectionInfo) {
        super(connectionInfo);
    }

    public RedshiftJdbcDialect(Connection connection) {
        super(connection);
    }

    @Override
    protected Driver getDriver() throws SQLException {
        return loadDriver(DataSourceType.REDSHIFT.getDriverClass());
    }

    @Override
    protected boolean filterSchema(String schemaName) {
        // redshift 只支持小写shcema和tablename
        if (StringUtils.isNotBlank(schemaName)) {
            schemaName = schemaName.toLowerCase(Locale.ROOT);
        }

        if (ArrayUtils.contains(SYS_SCHEMA_NAMES, schemaName)) {
            return false;
        }
        return super.filterSchema(schemaName);
    }

    @Override
    protected String getDDLColumnName(BeeTableType type) {
        return "Show Table DDL statement";
    }

    @Override
    public List<String> getDatabases() {
        try (Connection connection = this.getConnection();
                PreparedStatement preparedStatement = connection.prepareStatement(QUERY_DATABASES_SQL);
                ResultSet resultSet = preparedStatement.executeQuery()) {

            List<String> catalogs = Lists.newArrayList();
            while (resultSet.next()) {
                String catalogName = resultSet.getString(1);
                catalogs.add(catalogName);
            }
            return catalogs;
        } catch (SQLException e) {
            throw new RuntimeException("Get Scheme Error: " + e.getMessage(), e);
        }
    }

    @Override
    protected String createTableSql(BeeTableType type, String schemaName, String tableName) {
        // redshift 只支持小写shcema和tablename
        if (StringUtils.isNotBlank(schemaName)) {
            schemaName = schemaName.toLowerCase(Locale.ROOT);
        }
        if (StringUtils.isNotBlank(tableName)) {
            tableName = tableName.toLowerCase(Locale.ROOT);
        }

        if (BeeTableType.TABLE == type) {
            return "SHOW TABLE \"" + schemaName + "\".\"" + tableName + "\"";
        } else {
            return "SHOW VIEW \"" + schemaName + "\".\"" + tableName + "\"";
        }
    }

    @Override
    public List<MetaColumn> getSchemaColumns(String schemaName, String tableName) {
        // redshift 只支持小写shcema和tablename
        if (StringUtils.isNotBlank(schemaName)) {
            schemaName = schemaName.toLowerCase(Locale.ROOT);
        }
        if (StringUtils.isNotBlank(tableName)) {
            tableName = tableName.toLowerCase(Locale.ROOT);
        }
        return super.getSchemaColumns(schemaName, tableName);
    }

    @Override
    public MetaTable getSchemaTable(String schemaName, String tableName, boolean containColumn) {
        // redshift 只支持小写shcema和tablename
        if (StringUtils.isNotBlank(schemaName)) {
            schemaName = schemaName.toLowerCase(Locale.ROOT);
        }
        if (StringUtils.isNotBlank(tableName)) {
            tableName = tableName.toLowerCase(Locale.ROOT);
        }
        return super.getSchemaTable(schemaName, tableName, containColumn);
    }

    @Override
    public List<MetaTable> getSchemaTables(String schemaName) {
        if (StringUtils.isNotBlank(schemaName)) {
            schemaName = schemaName.toLowerCase(Locale.ROOT);
        }
        return super.getSchemaTables(schemaName);
    }

    @Override
    protected BeeTableType getTableType(String type) {
        if (StringUtils.equalsIgnoreCase(type, "v")) {
            return BeeTableType.VIEW;
        } else if (StringUtils.equalsIgnoreCase(type, "r")) {
            return BeeTableType.TABLE;
        } else if (StringUtils.equalsIgnoreCase(type, "m")) {
            return BeeTableType.MATERIALIZED_VIEW;
        } else {
            throw new IllegalArgumentException("table type is invalid: " + type);
        }
    }

    @Override
    protected String getSelectTablesSql(String tableName) {
        String sql = "SELECT n.nspname              AS TABLE_SCHEMA\n"
                + "     , c.relname              AS TABLE_NAME\n"
                + "     , c.relkind as TABLE_TYPE\n"
                + "     , null      AS TABLE_ENGINE\n"
                + "     , obj_description(c.oid) AS TABLE_COMMENT\n"
                + "     , c.reltuples AS TABLE_ROWS\n"
                + "     , cast(c.relpages as bigint) * 8 * 1024 as DATA_SIZE\n"
                + "     , (CASE WHEN c.relkind = 'p' THEN 'p' ELSE null END) as TABLE_PARTITION\n"
                + "     , null AS CREATE_TIME\n"
                + "     , null AS UPDATE_TIME\n"
                + "FROM pg_class c\n"
                + "         LEFT JOIN pg_namespace n\n"
                + "                   ON n.oid = c.relnamespace\n"
                + "WHERE c.relkind in ('r', 'f', 'p', 'v', 'm')\n"
                + "  AND n.nspname = :schemaName";

        if (StringUtils.isNotBlank(tableName)) {
            sql += " AND c.relnam = :tableName";
        }
        return sql;
    }

    @Override
    protected String getSelectColumnsSql() {
        return "SELECT\n"
                + "    a.attname AS column_name,\n"
                + "    t.typname AS DATA_TYPE,\n"
                + "    CASE\n"
                + "        WHEN a.atttypmod = -1 THEN t.typname\n"
                + "        WHEN t.typname = 'varchar' THEN t.typname || '(' || (a.atttypmod - 4) || ')'\n"
                + "        WHEN t.typname = 'bpchar' THEN 'char' || '(' || (a.atttypmod - 4) || ')'\n"
                + "        WHEN t.typname = 'numeric' OR t.typname = 'decimal' THEN t.typname || '(' || ((a.atttypmod - 4) >> 16) || ', ' || ((a.atttypmod - 4) & 65535) || ')'\n"
                + "        WHEN t.typname = 'bit' OR t.typname = 'bit varying' THEN t.typname || '(' || (a.atttypmod - 4) || ')'\n"
                + "        WHEN t.typname IN ('time', 'timetz', 'timestamp', 'timestamptz') THEN t.typname || '(' || a.atttypmod || ')'\n"
                + "        ELSE t.typname || ''\n"
                + "    END AS COLUMN_TYPE,\n"
                + "    CASE\n"
                + "        WHEN a.atttypmod = -1 THEN NULL\n"
                + "        WHEN t.typname IN ('varchar', 'bpchar', 'bit', 'bit varying') THEN a.atttypmod - 4\n"
                + "        WHEN t.typname IN ('numeric', 'decimal') THEN (a.atttypmod - 4) >> 16\n"
                + "        ELSE NULL\n"
                + "    END AS DATA_LENGTH,\n"
                + "    CASE\n"
                + "        WHEN a.atttypmod = -1 THEN NULL\n"
                + "        WHEN t.typname IN ('numeric', 'decimal') THEN (a.atttypmod - 4) & 65535\n"
                + "        WHEN t.typname IN ('time', 'timetz', 'timestamp', 'timestamptz') THEN a.atttypmod\n"
                + "        ELSE NULL\n"
                + "    END AS NUMERIC_SCALE,\n"
                + "    NULL as NUMERIC_PRECISION,\n"
                + "    null AS DATETIME_PRECISION,\n"
                + "    d.description AS COLUMN_COMMENT,\n"
                + "    pg_get_expr(ad.adbin, ad.adrelid) AS DEFAULT_VALUE,\n"
                + "    CASE WHEN a.attnotnull THEN 'NO' ELSE 'YES' END AS IS_NULLABLE\n"
                + "FROM\n"
                + "    pg_class c\n"
                + "    JOIN pg_namespace n ON c.relnamespace = n.oid\n"
                + "    JOIN pg_attribute a ON a.attrelid = c.oid\n"
                + "    JOIN pg_type t ON a.atttypid = t.oid\n"
                + "    LEFT JOIN pg_description d ON c.oid = d.objoid AND a.attnum = d.objsubid\n"
                + "    LEFT JOIN pg_attrdef ad ON a.attnum = ad.adnum AND a.attrelid = ad.adrelid\n"
                + "WHERE\n"
                + "    n.nspname = :schemaName\n"
                + "    AND c.relname = :tableName\n"
                + "    AND a.attnum > 0\n"
                + "ORDER BY\n"
                + "    a.attnum";
    }
}
